package reactor;

import org.testng.Assert;
import reactor.core.publisher.Flux;
import reactor.core.publisher.Mono;

import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.CompletableFuture;
import java.util.stream.Collectors;
import java.util.stream.Stream;

public class Demo1 {

    private void t1() {
        var future = CompletableFuture.supplyAsync(ArrayList::new)
                .thenApply(res -> {
                    res.add(1);
                    return res;
                });
    }

    private void t2() {

        CompletableFuture<List<String>> ids = ifhIds();

        CompletableFuture<List<String>> result = ids.thenComposeAsync(l -> {
            Stream<CompletableFuture<String>> zip = l.stream()
                    .map(i -> {

                        CompletableFuture<String> nameTask = ifhName(i);
                        CompletableFuture<Integer> statTask = ifhStat(i);

                        return nameTask.thenCombineAsync(statTask, (name, stat) -> "Name " + name + " has stats " + stat);
                    });

            List<CompletableFuture<String>> combinationList = zip.collect(Collectors.toList());
            CompletableFuture<String>[] combinationArray = combinationList.toArray(new CompletableFuture[combinationList.size()]);

            CompletableFuture<Void> allDone = CompletableFuture.allOf(combinationArray);

            return allDone.thenApply(v -> combinationList.stream()
                    .map(CompletableFuture::join)
                    .collect(Collectors.toList()));
        });

        List<String> results = result.join();

        Assert.assertTrue(results.containsAll(List.of( "Name NameJoe has stats 103",
                "Name NameBart has stats 104",
                "Name NameHenry has stats 105",
                "Name NameNicole has stats 106",
                "Name NameABSLAJNFOAJNFOANFANSF has stats 121")));

        Mono.empty();
    }

    /**
     * 使用Reactor改造CompletableFuture的回调
     */
    private void t3() {

        Flux<String> ids = ifhrIds(); // (1)

        // 从CompletableFuture获得Flux，Flux没有方法直接从Future获取Flux，需要通过Mono转换
//        ids = Mono.fromCompletionStage(ifhIds()).flatMapIterable(id -> id);

        Flux<String> combinations =
                ids.flatMap(id -> { // (2)
                    Mono<String> nameTask = ifhrName(id); // (3)
                    Mono<Integer> statTask = ifhrStat(id); // (4)

                    // Mono可以表示异步处理结果
//                    nameTask = Mono.fromCompletionStage(ifhName(id));
//                    statTask = Mono.fromCompletionStage(ifhStat(id));

                    return nameTask.zipWith(statTask, // (5)
                            (name, stat) -> "Name " + name + " has stats " + stat);
                });

        // Flux无法直接get所有结果，只能得到第一个或者最后一个结果，所以先把Flux转换为Mono，再调用阻塞方法得到计算结果
        Mono<List<String>> result = combinations.collectList(); // (6)

        List<String> results = result.block(); // (7)
        Assert.assertTrue(results.containsAll(List.of( // (8)
                "Name NameJoe has stats 103",
                "Name NameBart has stats 104",
                "Name NameHenry has stats 105",
                "Name NameNicole has stats 106",
                "Name NameABSLAJNFOAJNFOANFANSF has stats 121"
        )));
    }

    private Mono<Integer> ifhrStat(String id) {
        return Mono.empty();
    }

    private Mono<String> ifhrName(String id) {
        return Mono.empty();
    }

    private Flux<String> ifhrIds() {
        return Flux.empty();
    }

    private CompletableFuture<Integer> ifhStat(String i) {
        return CompletableFuture.supplyAsync(() -> 1);
    }

    private CompletableFuture<String> ifhName(String i) {
        return CompletableFuture.supplyAsync(() -> "a");
    }

    private CompletableFuture<List<String>> ifhIds() {
        return null;
    }
}
